<?php
namespace amqp;

use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Connection\AMQPStreamConnection;

class RabbitMQ
{
    protected $connection;
    protected $channel;
    protected $exchange;
    protected $routing;
    protected $queue;
    protected $config;
    const EXCHANGE_MODEL = 'fanout';                //交换机模式
    public function __construct($cache_exchange, $cache_queue)
    {
        $this->config=[
            'host'=>'localhost',
            'port'=>5672,
            'user'=>'guest',
            'password'=>'guest',
            'vhost'=>'/'
        ];
        $config = config('amqp');
        if($config){
            $this->config=array_merge($this->config,$config);
        }

        $this->connection = new AMQPStreamConnection($this->config['host'], $this->config['port'], $this->config['user'], $this->config['password'], $this->config['vhost']);

        if (!$this->connection->isConnected()) {
            throw  new \Exception('建立连接失败');
        }

        $this->channel = $this->connection->channel();

        if (!$this->channel->is_open()) {
            throw  new \Exception('通道连接失败');
        }

        if (!$cache_exchange) {
            throw  new \Exception('请设置交换机名称');
        } else {
            $this->exchange = $cache_exchange;
        }

        if (!$cache_queue) {
            throw  new \Exception('请设置队列名称');
        } else {
            $this->queue = $cache_queue;
            //路由 （同名 队列 借用）
            $this->routing = $this->queue;
        }
    }


    public function send($data,$ack=null,$failed=null)
    {
        $data=serialize($data);
        //1、声明 交换机
        $this->channel->exchange_declare($this->exchange, self::EXCHANGE_MODEL, false, true, false);

        //2、声明队列
        $this->channel->queue_declare($this->queue, false, true, false, false, false);

        //3、队列绑定 交换机
        $this->channel->queue_bind($this->queue, $this->exchange, $this->routing);

        //4.1 设置异步回调消息确认 （生产者 防止信息丢失）
        if($ack){
            $this->channel->set_ack_handler($ack);
        }
        if($failed){
            $this->channel->set_nack_handler($failed);
        }
        //4.2 选择为 confirm 模式（此模式不可以和事务模式 兼容）
        $this->channel->confirm_select();

        //5、发送消息 到队列
        $msg = new AMQPMessage($data, array(
            //参数 发送消息的时候将消息的 deliveryMode 设置为 2 （RabbitMQ 持久化 步骤二：消息）
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ));
        $this->channel->basic_publish($msg, $this->exchange, $this->queue);

        //4.3 阻塞等待消息确认 （生产者 防止信息丢失）
        $this->channel->wait_for_pending_acks();
        return true;
    }


    public function subscribe($callback)
    {
        //1、声明交换机
        $this->channel->exchange_declare($this->exchange, self::EXCHANGE_MODEL,false,true,false);
        //2、声明队列
        $this->channel->queue_declare($this->queue,false,true,false,false,false);
        //3、交换机和队列 绑定
        $this->channel->queue_bind($this->queue, $this->exchange,$this->routing);
        echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
        $func = function ($msg)use ($callback){
            echo '[x] Received :'.$msg->body."\n";
            $body=unserialize($msg->body);
             call_user_func_array($callback,['body'=>$body]);
            //手动 回复队列，message已消费
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };

        //只有consumer已经处理，并确认了上一条message时queue才分派新的message给它(非公平分配，如果存在耗时操作，那么也一直等待。现在是空闲领取消息)
        $this->channel->basic_qos(null, 1, null);
        //第四个参数  是否自动回应 ack，false 手动回应
        $this->channel->basic_consume($this->queue,'',false,false,false,false,$func);
        //进入等待状态
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
        return true;
    }



    /*
     * 关闭连接
     */
    public function __destruct()
    {
        // TODO: Implement __destruct() method.
        $this->channel->close();
        $this->connection->close();
    }

}
